package com.general.demo.bulkhead;

import io.github.resilience4j.bulkhead.*;
import io.vavr.CheckedFunction0;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;

import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.CompletionStage;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @description: BulkheadFactory
 * @author: general
 * @version: 1.0
 * @create: 2021-09-05 20:46
 **/
public class BulkheadFactory {

    private static final Logger logger = LoggerFactory.getLogger(BulkheadFactory.class);

    private static final String SEMAPHORE_BULKHEAD = "SemaphoreBulkhead";
    private static final String FIXED_THREAD_POOL_BULKHEAD = "FixedThreadPoolBulkhead";

    private static BulkheadFactory instance;

    public BulkheadFactory() {

    }

    public static BulkheadFactory getInstance(){
        if(instance==null){
            instance=new BulkheadFactory();
        }
        return instance;
    }

    /**
     * 返回一个bulkhead对象
     * @return
     */
    public Result getBulkheadResult(Invoker<?> invoker, Invocation invocation){
        Properties properties = new Properties();
        // 使用ClassLoader加载properties配置文件生成对应的输入流
        InputStream in = BulkheadFactory.class.getClassLoader().getResourceAsStream("bulkhead.properties");
        // 使用properties对象加载输入流
        try {
            properties.load(in);
            // 获取模式
            String mode = properties.getProperty("bulkhead.mode");
            if(StringUtils.isEmpty(mode)){
                return null;
            }
            if(SEMAPHORE_BULKHEAD.equals(mode)){
                return getSemaphoreResult(properties,invoker,  invocation);
            }else if(mode.equals(FIXED_THREAD_POOL_BULKHEAD)){
                return getThreadPoolResult(properties,  invoker,  invocation);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }

    private Result getThreadPoolResult(Properties properties,  Invoker<?> invoker, Invocation invocation) {
        String  maxThreadPoolSize = properties.getProperty("fixedThreadPoolBulkhead.maxThreadPoolSize",
                String.valueOf(Runtime.getRuntime()
                .availableProcessors()));
        String coreThreadPoolSize = properties.getProperty("fixedThreadPoolBulkhead.coreThreadPoolSize",
                String.valueOf(Runtime.getRuntime()
                        .availableProcessors() - 1));
        String queueCapacity = properties.getProperty("fixedThreadPoolBulkhead.queueCapacity", "100");
        String keepAliveDuration = properties.getProperty("fixedThreadPoolBulkhead.keepAliveDuration", "20");


        ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
                .maxThreadPoolSize(Integer.valueOf(maxThreadPoolSize))
                .coreThreadPoolSize(Integer.valueOf(coreThreadPoolSize))
                .queueCapacity(Integer.valueOf(queueCapacity))
                .keepAliveDuration(Duration.ofMillis(Integer.valueOf(keepAliveDuration)))
                .build();
        ThreadPoolBulkheadRegistry registry = ThreadPoolBulkheadRegistry.of(config);

        // Get or create a ThreadPoolBulkhead from the registry -
        // bulkhead will be backed by the default config
        ThreadPoolBulkhead bulkhead = registry.bulkhead(FIXED_THREAD_POOL_BULKHEAD);

        Supplier<Result> supplier = new Supplier<Result>() {
            @Override
            public Result get() {
                return invoker.invoke(invocation);
            }
        };

        final Result[] finalResult = {null};
        Supplier<CompletionStage<Result>> stageSupplier = bulkhead
                .decorateSupplier(supplier);
        stageSupplier.get().thenAccept(new Consumer<Result>() {
            @Override
            public void accept(Result result) {
                finalResult[0] = result;
            }
        });
        return finalResult[0];
    }

    private Result getSemaphoreResult(Properties properties, Invoker<?> invoker, Invocation invocation) {
        String  maxConcurrentCalls = properties.getProperty("semaphoreBulkhead.maxConcurrentCalls", "25");
        String maxWaitDuration = properties.getProperty("semaphoreBulkhead.maxWaitDuration", "0");
        BulkheadConfig config = BulkheadConfig.custom()
                .maxConcurrentCalls(Integer.valueOf(maxConcurrentCalls))
                .maxWaitDuration(Duration.ofMillis(Integer.valueOf(maxWaitDuration)))
                .build();
        BulkheadRegistry registry = BulkheadRegistry.of(config);
        Bulkhead bulkhead = registry.bulkhead(SEMAPHORE_BULKHEAD, config);
        CheckedFunction0<Result> decoratedSupplier = Bulkhead
                .decorateCheckedSupplier(bulkhead, () -> invoker.invoke(invocation));
        try {
            return decoratedSupplier.apply();
        } catch (Throwable throwable) {
            logger.error("getSemaphoreResult exception, e"+throwable);
            Result result = new AppResponse();
            result.setException(throwable);
            return result;
        }
    }


}
